Streamlit in Snowflake で時系列予測の結果を可視化してみた #SnowflakeDB

--データベース・スキーマの作成
USE ROLE SYSADMIN;
CREATE DATABASE streamlit_db;
CREATE SCHEMA streamlit_db.data;
CREATE SCHEMA streamlit_db.app;
CREATE SCHEMA streamlit_db.models;

--ウェアハウスの作成
CREATE WAREHOUSE streamlit_wh 
    WAREHOUSE_SIZE = small 
    WAREHOUSE_TYPE = standard
    AUTO_SUSPEND = 60
    AUTO_RESUME = True
    INITIALLY_SUSPENDED = True;

--roleの作成
USE ROLE USERADMIN;
CREATE ROLE streamlit_creator;

--権限の付与
USE ROLE SECURITYADMIN;
----ロール階層
GRANT ROLE streamlit_creator TO ROLE SYSADMIN;

----ウェアハウスの使用権限
GRANT USAGE ON WAREHOUSE streamlit_wh TO ROLE streamlit_creator;

----データアクセス：データ格納スキーマへの使用権限、テーブル・ビューの作成権限
GRANT USAGE ON SCHEMA streamlit_db.data TO ROLE streamlit_creator;
GRANT SELECT ON ALL TABLES IN SCHEMA streamlit_db.data TO ROLE streamlit_creator;
GRANT CREATE TABLE ON SCHEMA streamlit_db.data TO ROLE streamlit_creator;
GRANT CREATE VIEW ON SCHEMA streamlit_db.data TO ROLE streamlit_creator;

----streamlit権限：データベース・アプリ作成先のスキーマへの使用権限、スキーマレベルのStreamlitアプリ作成に必要な権限
GRANT USAGE ON SCHEMA streamlit_db.app TO ROLE streamlit_creator;
GRANT USAGE ON DATABASE streamlit_db TO ROLE streamlit_creator;
GRANT CREATE STREAMLIT ON SCHEMA streamlit_db.app TO ROLE streamlit_creator;
GRANT CREATE STAGE ON SCHEMA streamlit_db.app TO ROLE streamlit_creator;

----予測モデルの作成に必要なスキーマレベルの権限と作成先スキーマへの使用権限
GRANT CREATE SNOWFLAKE.ML.FORECAST ON SCHEMA streamlit_db.models TO ROLE streamlit_creator;
GRANT USAGE ON SCHEMA streamlit_db.models TO ROLE streamlit_creator;

--権限確認
SHOW GRANTS TO ROLE streamlit_creator;

/*============================================
分析
==============================================*/

--元データが日付単位なので、ここでは月単位に集約
USE ROLE STREAMLIT_CREATOR;
USE WAREHOUSE STREAMLIT_WH;
USE SCHEMA STREAMLIT_DB.DATA;

CREATE OR REPLACE TABLE monthly_sales AS
SELECT
    DATE_TRUNC('MONTH',  TO_TIMESTAMP(order_date, 'YYYY/MM/DD')) AS order_year_month
    ,SUM(sales) AS total_sales
FROM
    orders
GROUP BY
    DATE_TRUNC('MONTH',TO_TIMESTAMP(order_date, 'YYYY/MM/DD'))
ORDER BY
    order_year_month;

--データの期間を確認
SELECT order_year_month
FROM monthly_sales;

--model作成用に2016年までのデータを抽出
CREATE VIEW VIEW_MONTHLY_SALES AS 
    SELECT
        *
    FROM MONTHLY_SALES
    WHERE YEAR(order_year_month) <= 2016;

SELECT * FROM VIEW_MONTHLY_SALES;

-- モデルの構築
CREATE SNOWFLAKE.ML.FORECAST models.my_model(
    INPUT_DATA => SYSTEM$REFERENCE('VIEW', 'VIEW_MONTHLY_SALES'),
    TIMESTAMP_COLNAME => 'ORDER_YEAR_MONTH',
    TARGET_COLNAME => 'TOTAL_SALES'
);

--作成したモデルの確認
USE SCHEMA STREAMLIT_DB.MODELS;
SHOW SNOWFLAKE.ML.FORECAST;

--予測：1年間予測
CALL my_model!FORECAST(
    FORECASTING_PERIODS => 12,
    CONFIG_OBJECT => {'prediction_interval': 0.95}
    );
CREATE TABLE STREAMLIT_DB.DATA.My_forecasts AS SELECT * FROM TABLE(RESULT_SCAN(-1));

-- 予測結果
USE SCHEMA STREAMLIT_DB.DATA;
SELECT * FROM My_forecasts;

-- モデル作成用のデータと予測結果をユニオン
SELECT 
    ORDER_YEAR_MONTH
    ,TOTAL_SALES
    , 'Actual' AS type
    , NULL AS LOWER_BOUND
    , NULL AS UPPER_BOUND
FROM STREAMLIT_DB.DATA.VIEW_MONTHLY_SALES
UNION ALL
SELECT
    TS as ORDER_YEAR_MONTH
    ,FORECAST AS TOTAL_SALES 
    ,'Forecast' AS type
    ,LOWER_BOUND
    ,UPPER_BOUND
FROM STREAMLIT_DB.DATA.MY_FORECASTS;

# Import python packages
import streamlit as st
from snowflake.snowpark.context import get_active_session
import pandas as pd
from plotnine import ggplot, aes, geom_line, geom_point, geom_ribbon, labs, theme, element_text

session = get_active_session()

pd.set_option("max_colwidth",None)

# 関数定義

def create_model():
    cmd = f"""
        CREATE OR REPLACE SNOWFLAKE.ML.FORECAST STREAMLIT_DB.MODELS.MY_MODEL(
        INPUT_DATA => SYSTEM$REFERENCE('VIEW', 'STREAMLIT_DB.DATA.VIEW_MONTHLY_SALES'),
        TIMESTAMP_COLNAME => 'ORDER_YEAR_MONTH',
        TARGET_COLNAME => 'TOTAL_SALES'); 
    """
    res_df = session.sql(cmd).collect()
    return res_df 

def exe_forecast(input_period):
    cmd = f"""
         CREATE OR REPLACE TABLE STREAMLIT_DB.DATA.MY_FORECASTS AS
            SELECT * FROM TABLE(STREAMLIT_DB.MODELS.MY_MODEL!FORECAST(FORECASTING_PERIODS => {input_period}));
    """
    df_predicted = session.sql(cmd).collect()
    return df_predicted

def union_data():
    cmd = """
    SELECT 
    ORDER_YEAR_MONTH
        ,TOTAL_SALES
        ,'Actual' AS type
        ,NULL AS LOWER_BOUND
        ,NULL AS UPPER_BOUND
    FROM STREAMLIT_DB.DATA.VIEW_MONTHLY_SALES
    UNION ALL
    SELECT
        TS as ORDER_YEAR_MONTH
        ,FORECAST AS TOTAL_SALES 
        ,'Forecast' AS type
        ,LOWER_BOUND
        ,UPPER_BOUND
    FROM STREAMLIT_DB.DATA.MY_FORECASTS;
    """
    df_plot = session.sql(cmd).to_pandas()
    return df_plot

def get_forecasting_period():
    cmd = """
        SELECT
            left(date_trunc('MONTH',min(ts)),10) as min_year_month
            ,left(date_trunc('MONTH',max(ts)),10) as max_year_month
            ,COUNT(*) as cnt
        FROM STREAMLIT_DB.DATA.MY_FORECASTS;
    """
    df_period = session.sql(cmd).to_pandas()
    min_year_month = df_period['MIN_YEAR_MONTH'][0]
    max_year_month = df_period['MAX_YEAR_MONTH'][0]
    forecast_period_cnt = int(df_period['CNT'].iloc[0])

    return forecast_period_cnt,min_year_month,max_year_month

def create_plot(df):
    plot = (ggplot(df, aes(x='ORDER_YEAR_MONTH', y='TOTAL_SALES', color='TYPE'))
            + geom_line()
            + geom_point()
            + geom_ribbon(aes(ymin='LOWER_BOUND', ymax='UPPER_BOUND', fill='TYPE'), alpha=0.2,linetype='None')
            + labs(x='', y='TOTAL_SALES')
            + theme(axis_title_y=element_text(size=10),
                    axis_text=element_text(size=8),
                    axis_text_x=element_text(angle=90),
                    legend_title=element_text(size=8),
                    legend_text=element_text(size=8))
           )
    return plot

# 初期値の取得
if 'forecast_period_cnt' not in st.session_state:
    st.session_state.forecast_period_cnt, st.session_state.min_year_month, st.session_state.max_year_month = get_forecasting_period()

# サイドバーコンテンツ

with st.sidebar:
    st.write("■モデルを再構築")
    if st.button("Create Model"):
        with st.spinner('実行中です...'):
            res = create_model()
            st.success('Model creation successful!', icon="✅")
            st.write(res)

    st.write("■予測期間（月）を選択し [Execute Forecating] で結果を表示")
    input_period = st.number_input("予測期間（月）",min_value=1, max_value=24, value=st.session_state.forecast_period_cnt, step=1)

    if st.button("Execute Forecasting"):
        with st.spinner('実行中です...'):
            df_predicted = exe_forecast(input_period)
            st.success('Forecasting successful!', icon="✅")
            st.session_state.forecast_period_cnt, st.session_state.min_year_month, st.session_state.max_year_month = get_forecasting_period()
            st.write(df_predicted)    

# メインコンテンツ
st.title("Snowflake Cortex ML-based function Forecasting")
st.write("Snowflake Cortex ML-based function の一つである Forecasting による時系列の予測と予測結果の可視化")
st.write(f"■予測期間：{st.session_state.forecast_period_cnt} ヶ月（{st.session_state.min_year_month} ~ {st.session_state.max_year_month}）")

df = union_data()
plot = create_plot(df)

st.pyplot(plot.draw())
st.write(f"■予測データ")
st.dataframe(df)

USE ROLE USERADMIN;
CREATE ROLE streamlit_user;

--権限の付与
USE ROLE SECURITYADMIN;
----ロール階層
GRANT ROLE streamlit_user TO ROLE SYSADMIN;
GRANT USAGE ON DATABASE streamlit_db TO ROLE streamlit_user;
GRANT USAGE ON SCHEMA streamlit_db.app TO ROLE streamlit_user;
GRANT USAGE on STREAMLIT streamlit_db.app.U2UHUTC95R2VOFSA to ROLE streamlit_user;